package com.zhongfl.crm.lixiao.message;


import com.zhongfl.crm.model.boss.MessagePreventRepeat;
import com.zhongfl.crm.service.MessagePreventRepeatServiceApi;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.KafkaException;

import javax.annotation.Resource;


@Slf4j
public abstract class AbstractKafkaConsumer {

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Resource
    private MessagePreventRepeatServiceApi preventRepeatServiceApi;

    protected void handler(ConsumerRecord<String, String> data, String topic) {
        log.info("receive msg: {}", data);
        try {
            MessagePreventRepeat messagePreventRepeat = new MessagePreventRepeat();
            messagePreventRepeat.setGroupId(groupId);
            messagePreventRepeat.setMsgKey(data.key());
            messagePreventRepeat.setMsgValue(data.value());
            messagePreventRepeat.setTopic(topic);

            preventRepeatServiceApi.save(messagePreventRepeat);

            execute(data);
            
            log.info("execute success,Topic:{},Partition:{},Offset:{},TimesTamp:{}, Key:{},Value:{}"
                    , data.topic(), data.partition(), data.offset(), data.timestamp(), data.key(), data.value());
        } catch (DuplicateKeyException e) {
            log.info("message is executed,Topic:{},Partition:{},Offset:{},TimesTamp:{},Key:{},Value:{}"
                    , data.topic(), data.partition(), data.offset(), data.timestamp(), data.key(), data.value());
        } catch (KafkaException e) {
            log.error("message execute fail,Topic:{},Partition:{},Offset:{},TimesTamp:{},Key:{},Value:{}"
                    , data.topic(), data.partition(), data.offset(), data.timestamp(), data.key(), data.value(), e);
            throw e;
        } catch (Exception e) {
            log.error("message execute fail and don't know exception,Topic:{},Partition:{},Offset:{},TimesTamp:{},Key:{},Value:{}"
                    , data.topic(), data.partition(), data.offset(), data.timestamp(), data.key(), data.value(), e);
            throw new KafkaException("message execute fail", e);
        }
    }


    /**
     * 消息处理
     */
    public abstract void execute(ConsumerRecord<String, String> data) throws KafkaException;
}
